package com.shujia.streaming

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.streaming.dstream.DStream

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}
import scala.collection.mutable

object Demo03BuKong {
  def main(args: Array[String]): Unit = {

    /**
     * 缉查布控：根据黑名单对数据进行过滤
     */
    // 如果使用本地集合保存黑名单，那么如果需要增加黑名单用户则需要重启程序才能生效，代价比较高
    //    val blackListSet: Set[String] = Set[String](
    //      "D55433A437AEC8D8D3DB2BCA56E9E64392A9D93C"
    //      , "47BE1E866CFC071DB19D5E1C056BE28AE24C16E7"
    //      , "B6CA63F2DE56CA3BFBE8D0EC8632506AF7972556"
    //      , "82044609CB65BA7139205E7808144D5C254C419F"
    //      , "27A43BC5BB6853DE779A9EA7F3523733CACBD21C"
    //    )
    /**
     * 如果需要做到实时更新黑名单 并且不需要重启生效
     * 则可以选择一个外部存储系统 例如 MySQL来保存黑名单
     * 在MySQL中创建mdn_blacklist表
     * CREATE TABLE `mdn_blacklist` (
     * `mdn` varchar(255) NOT NULL,
     * PRIMARY KEY (`mdn`)
     * ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
     *
     * 插入数据：
     * insert into mdn_blacklist (mdn) values
     * ("D55433A437AEC8D8D3DB2BCA56E9E64392A9D93C")
     * , ("47BE1E866CFC071DB19D5E1C056BE28AE24C16E7")
     * , ("B6CA63F2DE56CA3BFBE8D0EC8632506AF7972556")
     * , ("82044609CB65BA7139205E7808144D5C254C419F")
     * , ("27A43BC5BB6853DE779A9EA7F3523733CACBD21C")
     */

    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo03BuKong")
    conf.setMaster("local[2]")
    val sc: SparkContext = new SparkContext(conf)
    // 每5s一个批次
    val ssc: StreamingContext = new StreamingContext(sc, Durations.seconds(5))

    // 每一条数据表示用户的位置记录
    val positionDS: DStream[String] = ssc.socketTextStream("master", 8888)

    // 需要结合黑名单数据 将黑名单中的 用户位置记录（经纬度） 过滤出来

    positionDS
      .filter(line => {
        // 如果连接的建立直接放在filter中会导致每一条数据会建立一次链接，链接建立的次数过多就会导致错误
        var flag: Boolean = false
        // 从基站数据中提取用户手机号 与 黑名单进行比对
        val mdn: String = line.split(",")(0)
        //        blackListSet.contains(mdn) // 使用本地集合无法实时更新黑名单
        // 从MySQL中获取黑名单数据
        val conn: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/student", "root", "123456")
        val pSt: PreparedStatement = conn.prepareStatement("select mdn from mdn_blacklist where mdn = ?")
        pSt.setString(1, mdn)
        val rs: ResultSet = pSt.executeQuery()
        var cnt: Int = 0
        while (rs.next()) {
          cnt += 1
        }
        println(cnt)
        if (cnt > 0) {
          flag = true
        }
        flag
      })
      // 提取数据中的位置信息
      .map(line => {
        val splits: Array[String] = line.split(",")
        val mdn: String = splits(0)
        val longitudeAndLatitude: String = splits(1)
        val longitude: String = longitudeAndLatitude.substring(0, 7)
        val latitude: String = longitudeAndLatitude.substring(7)
        val cityId: String = splits(2)
        val countyId: String = splits(3)
        s"$mdn,$cityId,$countyId,$longitude,$latitude"
      })
    //      .print()

    /**
     * filter算子会对DStream中的每一条数据依次进行处理
     * SparkStreaming程序运行时，每一个批次即每5s的数据会封装到一个Batch然后触发一个job进行处理
     * 实际上每一个Batch的数据会封装成一个RDD进行处理
     *
     * transform算子就可以将流处理转换为一个一个的Batch进行处理 即转换成一个一个的RDD进行处理
     */
    positionDS
      .transform(rdd => {
        // 如果需要使用DSL或者SQL进行处理 则可以将rdd转换成DF再进行操作 操作完之后再将DF转为RDD作为返回值

        // 从MySQL中获取黑名单数据 连接是不能被序列化的 所以无法放在算子外部
        val conn: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/student", "root", "123456")
        val pSt: PreparedStatement = conn.prepareStatement("select mdn from mdn_blacklist")

        // 将MySQL中的黑名单读取到本地 转成本地集合
        val rs: ResultSet = pSt.executeQuery()
        val set: mutable.HashSet[String] = mutable.HashSet[String]()
        while (rs.next()) {
          set.add(rs.getString("mdn"))
        }
        // 关闭连接
        rs.close()
        pSt.close()
        conn.close()

        // 广播变量
        val setBro: Broadcast[mutable.HashSet[String]] = sc.broadcast(set)

        val filterRDD: RDD[String] = rdd.filter(line => {
          // 从基站数据中提取用户手机号 与 黑名单进行比对
          val mdn: String = line.split(",")(0)
          setBro.value.contains(mdn)
        })
        filterRDD
      })
      // 提取数据中的位置信息
      .map(line => {
        val splits: Array[String] = line.split(",")
        val mdn: String = splits(0)
        val longitudeAndLatitude: String = splits(1)
        val longitude: String = longitudeAndLatitude.substring(0, 7)
        val latitude: String = longitudeAndLatitude.substring(7)
        val cityId: String = splits(2)
        val countyId: String = splits(3)
        s"$mdn,$cityId,$countyId,$longitude,$latitude"
      })
      .print()

    /**
     * foreachRDD vs transform
     * 都是将每个批次转换成RDD进行处理
     * 如果后续还需要处理数据 则选择transform
     * 如果是需要将数据写入到外部系统 则选择foreachRDD
     */
    //    positionDS.foreachRDD()

    ssc.start()
    ssc.awaitTermination()
    ssc.stop()


  }

}
